/**
 *    @file       netprocessthread.cpp
 *
 *    @brief      
 *    @details
 *
 *    @version    1.0
 *    @date       02/12/2014 05:47:43 PM
 *
 *    @author     Theophilus (), wangtf418@gmail.com 
 */
#include "netprocessthread.h"
#include "server.h"
#include "toolkit.h"
#define MAX_PROCMSGNUM_ONCE 20
#define CHECK_TASKTIMEOUT_INTERVAL 20
#define EPOLL_WAIT_SIZE 1000
#define EPOLL_WAIT_TIME 10

netprocessthread::netprocessthread(bf_uint32_t type, bf_uint32_t size, bf_uint32_t index, const comargs *args)
	: asynworkthread(type, size, index, args)
{
	m_last_checktime = toolkit::get_unixtime();
	m_check_interval = CHECK_TASKTIMEOUT_INTERVAL;
}

netprocessthread::~netprocessthread()
{

}

bf_int_t netprocessthread::init()
{
	bf_int_t ret = m_epoll.init(EPOLL_WAIT_SIZE, EPOLL_WAIT_TIME);
	if (ret != BF_OK)
	{
		return BF_ERROR;
	}
	return BF_OK;
}

bf_int_t netprocessthread::run()
{
	init_run();
	message *msg = NULL;
	bf_uint_t procnum = 0;
	while (m_runflag)
	{
		process_io_event();
		procnum = 0;
		while (procnum < MAX_PROCMSGNUM_ONCE)
		{
			msg = m_msg_queue.get();
			if (msg)
			{
				proc_msg(msg);
				delete msg;
			}
			else
			{
				break;
			}
			procnum ++;
		}
		check_task_timeout();
		process_other();
	}
	return BF_OK;
}

bf_int_t netprocessthread::process_io_event()
{
	bf_int_t ret = m_epoll.wait();
	if (ret != BF_TRUE)
	{
		return BF_OK;
	}
	m_epoll.process(*this);
	return BF_OK;
}

bf_int_t netprocessthread::process_other()
{
	return BF_OK;
}

bf_int_t netprocessthread::check_task_timeout()
{
	bf_time_t curtime = server::get()->get_time();
	if ((curtime - m_last_checktime) > m_check_interval)
	{
		std::map<bf_uint64_t, sockettask*>::iterator itr = m_tasks.begin();
		std::map<bf_uint64_t, sockettask*>::iterator itr_tmp;
		for (; itr != m_tasks.end();)
		{
			itr_tmp = itr ++;
			itr_tmp->second->check_timeout();
		}
		m_last_checktime = curtime;
	}
	return BF_OK;
}

bf_int_t netprocessthread::add_sockettask(sockettask *task)
{
	std::pair<sockettask_pos_t, bf_bool_t> pair_ret = m_tasks.insert(std::make_pair(task->get_taskid(), task));
	if (pair_ret.second)
	{
		task->set_pos(pair_ret.first);
		return BF_OK;
	}
	return BF_ERROR;
}

bf_int_t netprocessthread::del_sockettask(sockettask *task)
{
	m_tasks.erase(task->get_pos());
	return BF_OK;
}

sockettask * netprocessthread::find_sockettask(bf_uint64_t taskid)
{
	std::map<bf_uint64_t, sockettask*>::iterator itr = m_tasks.find(taskid);
	if (itr == m_tasks.end())
	{
		return NULL;
	}
	return itr->second;
}

void netprocessthread::process_epoll_event(bf_epoll_event_t &event)
{
	sockettask *task = (sockettask *)(event.data.ptr);
	if (task)
	{
		task->process_io_event(event.events);
	}
}

bf_int_t netprocessthread::proc_msg(message *msg)
{
	switch (msg->get_msghead().m_msgtype)
	{
		case MSG_TYPE_SEND:
			{
				proc_send_msg(msg);
				break;
			}
		case MSG_TYPE_RECV:
			{
				proc_recv_msg(msg);
				break;
			}
		case MSG_TYPE_SENDREQ:
			{
				proc_send_reqmsg(msg);
				break;
			}
		default:
			{
				break;
			}
	}
	return BF_OK;
}

bf_int_t netprocessthread::proc_recv_msg(message *msg)
{
	commsgbody<csocket> *body = (commsgbody<csocket>*)msg->get_msgbody();
	sockettask *task = server::get()->get_classfactory()->gen_sockettask(TSK_TYPE_TCP_SERVER, body->get_info(), this);
	if (task)
	{
		body->detach_info();
		task->init();
	}
	return BF_OK;
}

bf_int_t netprocessthread::proc_send_msg(message *msg)
{
	sockettask *task = find_sockettask(msg->get_msghead().m_tsk_id);
	if (task)
	{
		task->proc_msg(msg);
	}
	return BF_OK;
}

bf_int_t netprocessthread::proc_send_reqmsg(message *msg)
{
	msghead &head = msg->get_msghead();
	client_bufmsgbody *body = (client_bufmsgbody*)msg->get_msgbody();

	client_addr_args *args = new client_addr_args;
	args->m_srvip = body->get_srvip();
	args->m_srvport = body->get_srvport();
	args->m_sequence_num = body->get_sequence_num();

	args->m_thd_type = head.m_src_thd_type;
	args->m_thd_index = head.m_src_thd_index;
	args->m_tsk_id = head.m_src_tsk_id;
	sockettask *task = server::get()->get_classfactory()->gen_sockettask(TSK_TYPE_TCP_CLIENT, NULL, this);
	if (task)
	{
		task->init();
		task->put_send_packet(body->get_buf(), body->get_buflen());
		body->detach_buf();
	}
	else
	{
		delete args;
	}

	return BF_OK;
}

